{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 7 - Process Tasks in Order of Completion\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to use `ray.wait` to process tasks in the order that they finish.\n",
    "\n",
    "See the documentation for ray.wait at https://ray.readthedocs.io/en/latest/package-ref.html?highlight=ray.wait#ray.wait.\n",
    "\n",
    "## Concepts for this exercise - `ray.wait`\n",
    "\n",
    "After launching a number of tasks, you may want to run the results sequentially. To do so, we build off of exercise 6 and use `ray.wait` to execute the results sequentially. \n",
    "\n",
    "We are able to use `ray.wait` because the two lists returned by **`ray.wait` maintains the ordering of the input list**. That is, if `f` is a remote function, the code \n",
    "```python\n",
    "    results = ray.wait([f.remote(i) for i in range(100)], num_returns=10)\n",
    "```\n",
    "will return `(ready_list, remain_list)` and the `ObjectID`s of in those lists will be ordered by the argument passed to `f` above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=5, include_webui=False, ignore_reinit_error=True)\n",
    "\n",
    "# Sleep a little to improve the accuracy of the timing measurements used below,\n",
    "# because some workers may still be starting up in the background.\n",
    "time.sleep(2.0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def f():\n",
    "    time.sleep(np.random.uniform(0, 5))\n",
    "    return time.time()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Change the code below to use `ray.wait` to get the results of the tasks in the order that they complete.\n",
    "\n",
    "**NOTE:** It would be a simple modification to maintain a pool of 10 experiments and to start a new experiment whenever one finishes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start_time = time.time()\n",
    "\n",
    "remaining_result_ids = [f.remote() for _ in range(10)]\n",
    "\n",
    "# Get the results.\n",
    "results = []\n",
    "while len(remaining_result_ids) > 0:\n",
    "    # EXERCISE: Instead of simply waiting for the first result from\n",
    "    # remaining_result_ids, use ray.wait to get the first one to finish.\n",
    "    result_id = remaining_result_ids[0]\n",
    "    remaining_result_ids = remaining_result_ids[1:]\n",
    "    result = ray.get(result_id)\n",
    "    results.append(result)\n",
    "    print('Processing result which finished after {} seconds.'\n",
    "          .format(result - start_time))\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert results == sorted(results), ('The results were not processed in the '\n",
    "                                    'order that they finished.')\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
